[improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness #21498

poorbarcode · 2023-11-01T19:06:43Z

Motivation & Modifications

Add a new test used to verify that the Broker will not leave an orphan consumer in the scenario below:

Register "consumer-1"
- "consumer-1" will be maintained by the Subscription.
- "consumer-1" will be maintained by the Dispatcher.
The connection of "consumer-1" has something wrong. We call this connection "connection-1"
Try to register "consumer-2"
- "consumer-2" will be maintained by the Subscription. At this time, there are two consumers under this
  subscription.
- This will trigger a connection check task for connection-1, we call this task "CheckConnectionLiveness".
  This task will be executed in another thread, which means it will release the lock Synchronized(dispatcher)
- "consumer-2" was not maintained by the Dispatcher yet.
"CheckConnectionLiveness" will kick out "consumer-1" after 5 seconds, then "consumer-2" will be maintained
by the Dispatcher.

(Highlight) Race condition: if the connection of "consumer-2" went to a wrong state before step 4,
"consumer-2" is maintained by the Subscription and not maintained by the Dispatcher. Would the scenario below
will happen?

"connection-2" closed.
Remove "consumer-2" from the Subscription.
Try to remove "consumer-2" from the Dispatcher, but there are no consumers under this Dispatcher. To remove
nothing.
"CheckConnectionLiveness" is finished; put "consumer-2" into the Dispatcher.
At this moment, the consumer's state of Subscription and Dispatcher are not consistent. There is an orphan
consumer under the Dispatcher.

Why there is no orphan consumer?

There are two mechanisms to avoid orphan consumers:

After the consumer future is complete, it rechecks the connection is still active. Removed consumers again if the connection is inactive. see https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L919-L921
If the method consumerFuture.complet(v) returned false, broker will remove the consumer again. see https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L1271-L1285

PersistentTopic.java

if (!cnx.isActive()) {
    try {
        consumer.close();
    } 
    ...
}

ServerCnx.java

if (consumerFuture.complete(consumer)) {
    ...
} else {
    consumer.close();
}

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: x

lhotari

Great initiative to add a test for the described scenario. I provided some review feedback mainly about the new test scaffolding.

lhotari · 2023-11-02T12:48:19Z

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/InjectedClientCnxClientBuilder.java

+import org.apache.pulsar.common.api.proto.CommandWatchTopicUpdate;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+
+public class InjectedClientCnxClientBuilder {


Great idea to have a separate class for this.

lhotari · 2023-11-02T12:51:46Z

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/InjectedClientCnxClientBuilder.java

+        ClientConfigurationData conf = clientBuilder.getClientConfigurationData();
+        ThreadFactory threadFactory = new ExecutorProvider
+                .ExtendedThreadFactory("pulsar-client-io", Thread.currentThread().isDaemon());
+        EventLoopGroup eventLoopGroup =


This would currently leak threads since it doesn't get closed. One possibility would be to override the closeAsync and shutdown methods of PulsarClientImpl like it was done in #21468 for PulsarTestClient.

Fixed, I created a ulsarTestClient directly.

lhotari · 2023-11-02T13:01:48Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherLockTest.java

+            new ClientCnxCustomizer(Type.PING){
+                @Override
+                public void handleCommand(Object command) {
+                    if (command instanceof CommandPing) {
+                        // do not response anything.
+                    }
+                }
+            });


Is this ClientCnxCustomizer any easier than providing an actual anonymous ClientCnx class?

Passing a BiFunction<ClientConfigurationData, EventLoopGroup, ClientCnx> as a parameter in the InjectedClientCnxClientBuilder should be sufficient to support anything that is needed.

I think your suggestion is better.

I created a new implementation here, after the PR merged, I will back to finish this one.

Thanks

lhotari · 2023-11-02T13:07:10Z

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/InjectedClientCnxClientBuilder.java

+        return new PulsarClientImpl(conf, eventLoopGroup, pool);
+    }
+
+    public static abstract class ClientCnxCustomizer {


I don't see a reason to have ClientCnxCustomizer and InjectedClientCnx. Simply providing a BiFunction<ClientConfigurationData, EventLoopGroup, ClientCnx> parameter for creating the ClientCnx instance should be sufficient.
For example, you'd use it somewhat like this:

.clientCnxFunction((conf, eventLoopGroup) -> { return new ClientCnx(conf, eventLoopGroup) { @Override protected void handlePing(CommandPing ping) { // do not respond to CommandPing } }; });

The PulsarTestClient shows how it could be implemented. PulsarTestClient is specialized for a certain test, but it contains some useful ideas how clientCnxFunction could be used to instantiate the ClientCnx when it's needed.

lhotari · 2023-11-02T13:08:28Z

pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherLockTest.java

+        // Wait for all commands of the consumer c1 have been handled. To avoid the Broker mark the connection is active
+        // after it receive anything.
+        Thread.sleep(1000);


how do we know that 1000ms is sufficient?

…ondition occur

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Nov 1, 2023

poorbarcode changed the title ~~[improve] [test] Add a test to verify no orphan consumers if a race condition occur~~ [improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness Nov 1, 2023

poorbarcode self-assigned this Nov 1, 2023

poorbarcode requested review from lhotari and codelipenghui November 1, 2023 19:08

poorbarcode modified the milestones: 3.2.0, 3.0 Nov 1, 2023

poorbarcode added the release/3.0.3 label Nov 1, 2023

lhotari reviewed Nov 2, 2023

View reviewed changes

poorbarcode added 4 commits December 19, 2023 16:03

[improve] [test] Add a test to verify no orphan consumers if a race c…

1838b52

…ondition occur

[improve] [test] Add a test to verify no orphan consumers if a race c…

9045591

…ondition occur

cache

a6e6a3a

address comment

568431c

poorbarcode force-pushed the improve/test_orphan_consumer branch from b9a97a7 to 568431c Compare December 19, 2023 08:26

-

b647ed0

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

lhotari added release/3.0.8 and removed release/3.0.3 labels Oct 10, 2024

lhotari modified the milestones: 4.0.0, 4.1.0 Oct 11, 2024

lhotari added release/3.0.9 and removed release/3.0.8 labels Dec 6, 2024

lhotari added release/3.0.10 and removed release/3.0.9 labels Jan 3, 2025

lhotari added release/3.0.11 and removed release/3.0.10 labels Feb 19, 2025

lhotari added release/3.0.12 and removed release/3.0.11 labels Mar 31, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness #21498

[improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness #21498

poorbarcode commented Nov 1, 2023 •

edited

Loading

lhotari left a comment

lhotari Nov 2, 2023

lhotari Nov 2, 2023

poorbarcode Dec 19, 2023

lhotari Nov 2, 2023

poorbarcode Dec 8, 2023

poorbarcode Dec 19, 2023

lhotari Nov 2, 2023

lhotari Nov 2, 2023

[improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness #21498

Are you sure you want to change the base?

[improve] [test] Add a test to verify no orphan consumers if release lock caused by checkConnectionLiveness #21498

Conversation

poorbarcode commented Nov 1, 2023 • edited Loading

Motivation & Modifications

Why there is no orphan consumer?

Documentation

Matching PR in forked repository

lhotari left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

poorbarcode commented Nov 1, 2023 •

edited

Loading